package utils

import java.sql.DriverManager
import java.util

import org.apache.kafka.common.TopicPartition
import scalikejdbc.config.DBs
import scalikejdbc.{DB, SQL}
import scala.collection.mutable
/**
  * hashes
  * key:topic-groupId
  *     field: partition value:untilOffset
  *     field: partition value:untilOffset
  *     field: partition value:untilOffset
  *     field: partition value:untilOffset
  *     field: partition value:untilOffset
  *     field: partition value:untilOffset
  * Created by Administrator on 2018/03/25.
  */
object OffsetManager {


  def apply(topics: Array[String], groupId: String): Map[TopicPartition, Long] = {

    // 去redis中，把你之前存储的偏移量拿出来并封装成一个map对象即可
    val jedis = Jedis.getJedis
    // 通过key去获取对应的数据
    val redisMap: util.Map[String, String] = jedis.hgetAll(topics.head.concat("-").concat(groupId))
    val scalaMap: mutable.Map[String, String] = collection.JavaConversions.mapAsScalaMap(redisMap)
    jedis.close()

    scalaMap.map(tp => (new TopicPartition(topics.head, tp._1.toInt), tp._2.toLong)).toMap
  }
}
